# -*- coding: utf-8 -*-
"""
Created on Wed Jul  3 10:15:15 2019

@author: YUYH2
"""

import pymysql
import pandas as pd
import re
from sqlalchemy import create_engine
import config
import logging
import errorcode
#from pyhive import hive
from impala.dbapi import connect
from impala.util import as_pandas
import pyMySQLUtil
import dataupload

'''
连接mysql数据源
'''

class Dao:
    def __init__(self, host=config.mysql_host,port = config.mysql_port,user = config.mysql_user,password =config.mysql_password,database = config.mysql_db):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database

    """
    连接mysql
    """
    def connet_to_mysql(self):
        try:
            conn = pymysql.connect(host = self.host, port = self.port, user = self.user, password = self.password, database = self.database)
            return conn
        except Exception as e:
            logging.error(errorcode.Connect_Database_Error + ':' + str(e))
            raise e

    """
    连接hive
    """
    def connect_to_hive(self):
        conn = None
        try:
            conn = connect(host=self.host, port=self.port, auth_mechanism='PLAIN', user=self.user, password=self.password, database=self.database)
            return conn
        except Exception as e:
            print('hive连接错误: ' + str(e))
            return conn

    def query_databases_hive(self):
        """
        返回当前连接下的所有数据库
        :return: pd.DataFrame
        """
        try:
            #0 连接hive
            con = self.connect_to_hive()

            #1 查询sql语句
            sql = 'show databases'

            #2 执行sql语句
            cursor = con.cursor()
            cursor.execute(sql)
            data = as_pandas(cursor)

            #3 关闭连接
            con.close()
        except Exception as e:
            print('hive获取数据库信息失败: ' + str(e))

        return data

    def query_tables_hive(self):
        """
        返回当前连接的库的所有表
        :return: pd.DataFrame
        """
        try:
            #0 连接hive
            con = self.connect_to_hive()

            #1 查询sql语句
            sql = 'show tables'

            #2 执行sql语句
            cursor = con.cursor()
            cursor.execute(sql)
            data = as_pandas(cursor)

            #3 关闭连接
            con.close()
        except Exception as e:
            print('hive获取数据库信息失败: ' + str(e))

        return data


    """
    查询数据:mysql
    返回DataFrame
    :param columns:类型[],需要查询的列
    :param table_name:要查询的表，如果此参数为空，则自定义查询
    :param sql:查询的sql，如果此参数不为*，则查询获取指定指定字段
    """
    def query_data(self, table_name = None, sql = None, columns = None, constraint=None):
        
        #连接数据库
        conn = self.connet_to_mysql()
        cursor = conn.cursor(cursor=pyMySQLUtil.DictCursor)
        data = None
        
        """
        指定表查询
        """
        if not sql:
            # 查询全表数据
            if table_name != None and columns == None:
                sql = 'select * from {table_name} '.format(sql=sql,table_name = table_name)

            #查询指定字段
            elif table_name != None and columns != None:
                if isinstance(columns, list):
                    col = ','.join(columns)
                elif isinstance(columns, str):
                    col = columns
                else:
                    raise(Exception('列名参数错误，仅支持[]或者str型'))
                sql = 'select {col} from {table_name}'.format(col = col, table_name = table_name)

            #限制数据库数据查询
            if constraint:
                sql = sql + ' limit {}'.format(constraint)
            else:
                sql = sql + ' limit {}'.format(100000)
        else:
            sql = sql

        """
        执行sql语句
        """
        #print(sql)
        cursor.execute(sql)
        data_source = cursor.fetchall()
        
        #将数据转成数据框
        try:
            data = pd.DataFrame(data_source)
        except Exception as e:
            print('数据为空:', str(e))
            return data
        conn.close()
        return data

    """
    查询hive表
    """
    def query_data_hive(self, table_name=None, sql=None, columns=None, constraint=None):
        data = None

        # 连接hive
        con = self.connect_to_hive()

        try:
            if not sql:
                # 构造sql
                if table_name != None and columns == None:
                    sql = 'select * from {table_name}'.format(table_name=table_name)
                elif table_name != None and columns != None:
                    if isinstance(columns, list):
                        col = ','.join(columns)
                    elif isinstance(columns, str):
                        col = columns
                    else:
                        raise (Exception('列名参数错误，仅支持[]或者str型'))
                    sql = 'select {col} from {table_name}'.format(col=col, table_name=table_name)

                # 限制数据库数据查询
                if constraint:
                    sql = sql + ' limit {}'.format(constraint)
                else:
                    sql = sql + ' limit {}'.format(100000)
            else:
                sql = dataupload.remove_sign(sql) #  去掉sql语句末尾的;

            #执行sql
            cursor = con.cursor()
            cursor.execute(sql)
            data = as_pandas(cursor)
        except Exception as e:
            print('数据获取失败: ' + str(e))

        #关闭连接
        con.close()
        return data

    """
    执行hive语句
    """
    def excute_sql_hive(self, sql):
        sql = dataupload.remove_sign(sql)
        try:
            # 连接hive
            con = self.connect_to_hive()

            # 执行sql
            cursor = con.cursor()
            cursor.execute(sql)
            data = as_pandas(cursor)
        except Exception as e:
            print('数据获取失败: ' + str(e))

        con.close()
        return data

    """
    更新数据
    :param sql:需要执行的sql语句
    """
    def excute_sql(self, sql):
        try:
        #连接数据库
            conn = self.connet_to_mysql()
            cursor = conn.cursor()
            result = cursor.execute(sql)
            return result,cursor,conn
        except Exception as e:
            logging.error(errorcode.Excute_Sql_Error + ':' + str(e))
            raise e
    
    """
    删除数据
    支持两种删除操作，（1）删除全表；（2）执行sql语句删除
    :param table_name：表名，如果不为空，则删除全表
    :param sql:sql语句
    """
    def delete_data(self, table_name = None, sql = None):
        conn = self.connet_to_mysql()
        cursor = conn.cursor()
        if table_name is None and sql is not None:
            sql = sql
        elif table_name is not None and sql is None:
            sql = 'delete from {table_name}'.format(table_name = table_name)
        else:
            print('参数错误')
            return
        cursor.execute(sql)
        conn.commit()
        conn.close()
        
    """
    写入数据
    :param data:data,需要写入表的数据,类型DataFrame
    :param table_name:表名,需要写入表的表名
    """
    def insert_data(self, data, table_name):
        try:
            engine = create_engine('mysql+pymysql://' + self.user + ':' + str(self.password) + '@' + str(self.host) + ':' + str(self.port) +  '/' + self.database + '?charset=utf8')
            data.to_sql(table_name, engine, if_exists='append', index = False)
        except Exception as e:
            print(str(e))
            print(table_name,'写入数据失败')
    

    """
    获取本地数据
    :file:为文件路径及文件名，目前支持.xlsx .csv .txt
    :sheet:sheet名，仅当读取excel生效
    """
    @staticmethod
    def get_local_data(file, sheet = None):
        data = None
        try:
            if re.search('\.xlsx?$', file):
                if sheet is not None:
                    data = pd.read_excel(file, sheetname = sheet)
                else:
                    data = pd.read_excel(file)
            elif re.search('\.csv$', file):
                data = pd.read_csv(file)
            elif re.search('\.txt', file):
                data = pd.read_table(file)
            else:
                print('参数错误：不支持的文件')
        except Exception as e:
            print('读取文件错误: ' + str(e))
        return data
    
    """
    写入本地,支持结构化的数据(DataFrame), 写入csv、xlsx、txt
    :data:需要写入表的数据,如果要写入多份数据到excel,类型需要为[],元素为DataFrame类型
    :sheets:sheet名，写入excel失效,如果是写入多份数据到excel, sheets需要为[]
    """
    @staticmethod
    def write_to_local(file, data, sheets = None):
        try:
            if re.search('\.xlsx?$', file) and not isinstance(data, list):
                data.to_excel(file, sheets)
            elif re.search('\.csv$', file) and not isinstance(data, list):
                data.to_csv(file)
            elif re.search('\.xlsx?$', file) and isinstance(data, list) and isinstance(sheets, list):
                writer = pd.ExcelWriter(file)
                for i in range(len(data)):
                    data[i].to_excel(writer, sheets[i])
                writer.save()
                writer.close()
            elif re.search('\.txt?$', file):
                with open(file, mode='w', encoding='utf-8') as f:
                    f.writelines(str(data))        
        except Exception as e:
            print('写入数据失败: ' + str(e))

    # 根据字典插入数据
    def insert_from_dict(self,table_name, values_map):

        try:
            keys = ", ".join(values_map.keys())
            qmark = ", ".join(["%s"] * len(values_map))
            sql = "insert into {table_name}(%s) values (%s)".format(table_name=table_name) % (keys, qmark)
            conn = self.connet_to_mysql()
            cursor = conn.cursor()
            nums = cursor.execute(sql, list(values_map.values()))
            conn.commit()
            conn.close()
            return nums
        except Exception as e:
            logging.error(errorcode.Insert_Sql_Error + ':' + str(e))
            raise e


    # values_map:需要更新的key_value对，key是数据库字段，value是要更新的值
    # limit_map是查询条件：key是数据库字段，value是查询的值
    def update_from_dict(self,table_name,values_map,limit_map):

        try:
            conn = self.connet_to_mysql()
            cursor = conn.cursor()
            limit_keys = "=%s and ".join(limit_map.keys()) + '=%s'
            value_key = "=%s , ".join(values_map.keys()) + '=%s'
            all_values = list(values_map.values()) + list(limit_map.values())
            sql = "update {table_name} set %s where %s ".format(table_name=table_name) % (value_key, limit_keys)
            nums = cursor.execute(sql, all_values)
            conn.commit()
            conn.close()
            return nums
        except Exception as e:
            logging.error(errorcode.Update_Sql_Error + ':' + str(e))
            raise e

    def get_sql_list(self,sql):
        try:
            conn = self.connet_to_mysql()
            cursor = conn.cursor()
            cursor.execute(sql)
            result = cursor.fetchall()
            cursor.close()
            conn.close()
            return result
        except Exception as e:
            logging.error(errorcode.Query_Sql_Error + ':' + str(e))
            raise e
if __name__ == '__main__':
    # test1
    # dao = Dao()
    # sql = 'select * from training_model'
    # data = dao.query_data(sql = sql)
    # dao.insert_data(data, 'training_model_data')
    #
    # sql1 = 'select * from training_task'
    # data1 = dao.query_data(sql = sql1)
    # dao.insert_data(data1, 'training_task_data')

    # test hive测试：1、sql语句；2、tablename
    # dao_hive = Dao(host='10.133.106.107', port=10000, user='hive', database='meicloud_dpai_data_prd',
    #                password='mA8QwgfRRU')
    # sql = 'select * from t_datalake_device_params_sample_daily limit 1000'
    # data = dao_hive.excute_sql_hive(sql)
    # data = dao_hive.query_data_hive(sql = sql)
    # data = dao_hive.query_data_hive(table_name='t_datalake_device_params_sample_daily', constraint=100)
    #
    # # test mysql测试：1、sql语句；2、tablename
    # dao = Dao()
    # sql = 'select * from 3867iris limit 100'
    # data = dao.query_data(table_name='3867iris', constraint=50)


    # test hive测试
    dao_hive = Dao(host='10.16.74.31', port=10005, user='hive', database='temp',
                   password='hive123')
    sql = 'select * from temp_iris limit 100;'
    # data = dao_hive.excute_sql_hive(sql)
    data = dao_hive.query_data_hive(sql=sql)
    # print(data)
    # dao = Dao()
    # sql = 'select * from feature_engineering a ,test_table_1 b where a.id = b.id'
    # data = dao.query_data(sql = sql)
    # print(data)
    #mysql测试
    # dao = Dao()
    # data = dao.query_data(table_name='training_task', columns = ['create_time', 'arithmetic'], constraint=10)
    # print(data)

    #hive测试dao_hive.connect_to_hive()
    # dao_hive = Dao(host= '10.133.106.107', port=10000, user='hive', database='meicloud_dpai_data_chure_prd')
    # #sql = 'show tables'
    # sql = 'select * from t_datalake_device_params_sample_full'
    # #data = dao_hive.excute_sql_hive(sql)
    # data = dao_hive.query_data_hive(sql=sql)
    # print(data)


    #1
    #dao_hive = Dao(host='10.18.25.92', port=20001, user='datamining_bind', database='dm', password='mA8QwgfRRU')
    # sql = 'show tables'
    # data = dao_hive.excute_sql_hive(sql)

    #2
    # dao_hive = Dao(host='10.16.74.32', port=10000, user='fbd', database='default', password='F1b2d3$#')
    #
    # #test1 返回数据库
    # data = dao_hive.query_databases_hive()
    # print(data)
    # #test2 返回表
    # data = dao_hive.query_tables_hive()
    # print(data.head())
    # #test3 返回表所有数据
    # data = dao_hive.query_data('t_partner_plat_mapping_stg')
    # print(data.head())
    # #test4 执行sql
    # data = dao_hive.excute_sql_hive('show databases')
    # print(data)